--------------------------sql任务说明----------------------------------
-- @author :liuxiaojie05
-- @descrbe :职位发布流水 仅收集发布之后的日志 ods --->dwd
-- "operationType", "操作类型:提交页面,save接口,成功页"
-- operationType=save 这个字段等于save才是新增和修改；view是发布页 并没有产生提交表单的操作
-- logid字段可以串起来 把发布页  save 和 成功页 串起来
-- @udf jar包 : bigdata_realtime_flink_sql_udf-1.0-testlxj-jar-with-dependencies.jar

--------------------------sql任务说明----------------------------------

-- 加载函数
CREATE FUNCTION ZpDWDJobPublishV2Udtf as 'udtf.etl.dwd.ZpDWDJobPublishV2Udtf';
create function SplitFunction AS 'udtf.SplitFunction';

-- source
create table ods_zp_job_publis_test_v3 (
    log string
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_zhaopin_zppostplatform',
'properties.bootstrap.servers' = '10.135.9.4:9092,10.135.9.5:9092,10.135.9.6:9092,10.135.9.7:9092,10.135.9.8:9092',
'properties.group.id' = 'dwd_zp_job_publish_v2_202101141503',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_zhaopin_zppostplatform-Z60mN',
'format' = 'raw'
);

-- sink
create table dwd_zp_job_publis_test_v3 (
    infoId  string,
    entId   string,
    postSource string,
    log_key    string,
    dispCateId string,
    userAgent  string,
    isAdd      string,
    userId     string,
    entType    string,
    clientSource      string,
    clientType        string,
    entCompanySize    string,
    xxzl_deviceid     string,
    xxzl_smartid      string,
    isPostSuccess  string,
    xxzl_cid       string,
    entSize        string,
    userIP         string,
    postFailedMsg  string,
    operationType  string,
    logId        string,
    dispLocalId  string,
    sink_time    string,
    e_time       string
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_dwd_job_publish_test_v3',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin_dwd_zp_job_publish-sql-prod',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_job_publish_test_v3-WQvK6',
'format' = 'json'
);


-- sql-etl
insert into dwd_zp_job_publis_test_v3
select pp.*
from (
    select *
    from ods_zp_job_publis_test_v3, LATERAL TABLE(SplitFunction(log,'\n'))
) as tmp, LATERAL TABLE(ZpDWDJobPublishV2Udtf(regex_str)) as pp;
